package com.xwintop.xTransfer.sender.service.impl;

import com.google.common.base.Strings;
import com.xwintop.xTransfer.common.MsgLogger;
import com.xwintop.xTransfer.common.model.LOGKEYS;
import com.xwintop.xTransfer.common.model.LOGVALUES;
import com.xwintop.xTransfer.common.model.Msg;
import com.xwintop.xTransfer.messaging.IMessage;
import com.xwintop.xTransfer.sender.bean.SenderConfig;
import com.xwintop.xTransfer.sender.bean.SenderConfigPulsar;
import com.xwintop.xTransfer.sender.service.Sender;
import com.xwintop.xTransfer.task.quartz.TaskQuartzJob;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.client.api.*;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Service;

import java.util.HashSet;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * @ClassName: SenderPulsarImpl
 * @Description: Pulsar发送类
 * @author: xufeng
 * @date: 2018/12/28 17:04
 */

@Service("senderPulsar")
@Scope("prototype")
@Getter
@Setter
@Slf4j
public class SenderPulsarImpl implements Sender {
    private SenderConfigPulsar senderConfigPulsar;

    private Producer producer = null;
    private PulsarClient pulsarClient = null;

    @Override
    public Boolean send(IMessage msg, Map params) throws Exception {
        this.initPulsarClient();
        this.initProducer();
        try {
            TypedMessageBuilder messageBuilder = producer.newMessage();
            messageBuilder.key(msg.getId());
            messageBuilder.value(msg.getMessage());
            if (senderConfigPulsar.getArgs() != null && !senderConfigPulsar.getArgs().isEmpty()) {
                messageBuilder.properties(senderConfigPulsar.getArgs());
            }
            for (Map.Entry<Object, Object> objectObjectEntry : msg.getProperties().entrySet()) {
                Object value = objectObjectEntry.getValue();
                if (value instanceof String || value instanceof Number || value instanceof Boolean) {
                    messageBuilder.property((String) objectObjectEntry.getKey(), String.valueOf(value));
                }
            }
            if (StringUtils.isNotEmpty(senderConfigPulsar.getFileNameField())) {
                messageBuilder.property(senderConfigPulsar.getFileNameField(), msg.getFileName());
            }
            messageBuilder.send();

            Msg msgLogInfo = new Msg(LOGVALUES.EVENT_MSG_SENDED, msg.getId(), null);
            msgLogInfo.put(LOGKEYS.CHANNEL_IN_TYPE, msg.getProperty(LOGKEYS.CHANNEL_IN_TYPE));
            msgLogInfo.put(LOGKEYS.CHANNEL_IN, msg.getProperty(LOGKEYS.CHANNEL_IN));
            msgLogInfo.put(LOGKEYS.CHANNEL_OUT_TYPE, LOGVALUES.CHANNEL_TYPE_PULSAR);
            msgLogInfo.put(LOGKEYS.CHANNEL_OUT, senderConfigPulsar.getServiceUrl() + "/" + senderConfigPulsar.getTopic());
            msgLogInfo.put(LOGKEYS.MSG_TAG, msg.getFileName());
            msgLogInfo.put(LOGKEYS.MSG_LENGTH, ArrayUtils.getLength(msg.getMessage()));
            msgLogInfo.put(LOGKEYS.JOB_ID, params.get(TaskQuartzJob.JOBID));
            msgLogInfo.put(LOGKEYS.JOB_SEQ, params.get(TaskQuartzJob.JOBSEQ));
            msgLogInfo.put(LOGKEYS.RECEIVER_TYPE, msg.getProperty(LOGKEYS.RECEIVER_TYPE));
            msgLogInfo.put(LOGKEYS.RECEIVER_ID, msg.getProperty(LOGKEYS.RECEIVER_ID));
            MsgLogger.info(msgLogInfo.toMap());
        } catch (Exception e) {
            log.error("SendPulsar失败：", e);
            this.destroy();
            throw e;
        }
        return true;
    }

    @Override
    public void setSenderConfig(SenderConfig senderConfig) throws Exception {
        this.senderConfigPulsar = (SenderConfigPulsar) senderConfig;
    }

    @Override
    public void destroy() throws Exception {
        if (producer != null) {
            producer.close();
            producer = null;
        }
        if (pulsarClient != null) {
            pulsarClient.close();
            pulsarClient = null;
        }
    }

    public synchronized void initPulsarClient() throws PulsarClientException {
        if (pulsarClient == null) {
            ClientBuilder pulsarClientBuilder = PulsarClient.builder();
            pulsarClientBuilder.serviceUrl(senderConfigPulsar.getServiceUrl());
            pulsarClientBuilder.ioThreads(senderConfigPulsar.getIoThreads());
            pulsarClientBuilder.listenerThreads(senderConfigPulsar.getListenerThreads());
            pulsarClientBuilder.enableTcpNoDelay(senderConfigPulsar.isEnableTcpNoDelay());
            pulsarClientBuilder.keepAliveInterval(senderConfigPulsar.getKeepAliveIntervalSec(), TimeUnit.SECONDS);
            pulsarClientBuilder.connectionTimeout(senderConfigPulsar.getConnectionTimeoutSec(), TimeUnit.SECONDS);
            pulsarClientBuilder.operationTimeout(senderConfigPulsar.getOperationTimeoutSec(), TimeUnit.SECONDS);
            pulsarClientBuilder.startingBackoffInterval(senderConfigPulsar.getStartingBackoffIntervalMs(), TimeUnit.MILLISECONDS);
            pulsarClientBuilder.maxBackoffInterval(senderConfigPulsar.getMaxBackoffIntervalSec(), TimeUnit.SECONDS);
            pulsarClientBuilder.useKeyStoreTls(senderConfigPulsar.isUseKeyStoreTls());
            pulsarClientBuilder.tlsTrustCertsFilePath(senderConfigPulsar.getTlsTrustCertsFilePath());
            pulsarClientBuilder.tlsCiphers(new HashSet<>(senderConfigPulsar.getTlsCiphers()));
            pulsarClientBuilder.tlsProtocols(new HashSet<>(senderConfigPulsar.getTlsProtocols()));
            pulsarClientBuilder.tlsTrustStorePassword(senderConfigPulsar.getTlsTrustStorePassword());
            pulsarClientBuilder.tlsTrustStorePath(senderConfigPulsar.getTlsTrustStorePath());
            pulsarClientBuilder.tlsTrustStoreType(senderConfigPulsar.getTlsTrustStoreType());
            pulsarClientBuilder.allowTlsInsecureConnection(senderConfigPulsar.isAllowTlsInsecureConnection());
            pulsarClientBuilder.enableTlsHostnameVerification(senderConfigPulsar.isEnableTlsHostnameVerification());
            if (!Strings.isNullOrEmpty(senderConfigPulsar.getTlsAuthCertFilePath()) && !Strings.isNullOrEmpty(senderConfigPulsar.getTlsAuthKeyFilePath())) {
                pulsarClientBuilder.authentication(AuthenticationFactory.TLS(senderConfigPulsar.getTlsAuthCertFilePath(), senderConfigPulsar.getTlsAuthKeyFilePath()));
            }

            if (!Strings.isNullOrEmpty(senderConfigPulsar.getTokenAuthValue())) {
                pulsarClientBuilder.authentication(AuthenticationFactory.token(senderConfigPulsar.getTokenAuthValue()));
            }
            pulsarClient = pulsarClientBuilder.build();
        }
    }

    public synchronized void initProducer() throws Exception {
        if (producer == null) {
            ProducerBuilder producerBuilder = this.pulsarClient.newProducer(Schema.BYTES);
            producerBuilder.topic("persistent://" + senderConfigPulsar.getTenant() + "/" + senderConfigPulsar.getNamespace() + "/" + senderConfigPulsar.getTopic());
            producer = producerBuilder.create();
        }
    }
}
